package net.bwie.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author: 🌹帅的一塌糊涂🌹
 * @description:
 * @params: 分流判断 
 * @return:  
 * @date: 2025/5/18 21:00
 */
public class LogSplitProcessFuncion extends ProcessFunction<String, String> {
    private final OutputTag<String> errorTag ;
    private final OutputTag<String> strartTag ;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;
    public LogSplitProcessFuncion(OutputTag<String> errorTag, OutputTag<String> strartTag,
                                  OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.strartTag = strartTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        // TODO: 2025/5/18 解析json文件
        JSONObject jsonObject = JSON.parseObject(value);
        // TODO: 2025/5/18 错误日志 
        if (null != jsonObject.get("err")){
            // TODO: 2025/5/18 输出到日志测流
            ctx.output(errorTag,value);
            // TODO: 2025/5/18 防止后面主流还有错误日志 
            jsonObject.remove("err");
        }
        // TODO: 2025/5/18 启动日志
        if (null  != jsonObject.get("start")){
            // TODO: 2025/5/18  输出到启动日志测流
            ctx.output(strartTag,jsonObject.toJSONString());
        }else {
            // TODO: 2025/5/18 获取曝光和动作日志 
            JSONArray displays = jsonObject.getJSONArray("displays");
            JSONArray actions = jsonObject.getJSONArray("actions");
            // TODO: 2025/5/18 删除主流的曝光和动作日志
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            // TODO: 2025/5/18 输出页面日志
            out.collect(jsonObject.toJSONString());

            // TODO: 2025/5/18 判断曝光不等于空的数据
            if (null != displays && !displays.isEmpty()){
                for (int i = 0; i < displays.size(); i++){
                    JSONObject display = displays.getJSONObject(i);
                    jsonObject.put("display",display);
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }
            jsonObject.remove("display");

            // TODO: 2025/5/18 判断日志不等于空的数据
            if (null != actions && !actions.isEmpty()){
                for (int i = 0; i < actions.size(); i++){
                    JSONObject action = actions.getJSONObject(i);
                    jsonObject.put("action",action);
                    ctx.output(actionTag,jsonObject.toJSONString());
                }
            }

        }

    }
}
